package com.example.window;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;

import java.text.SimpleDateFormat;
import java.util.Random;

/**
 * 滑动窗口
 *
 * @author lbq
 * @date 2022/3/24 22:43
 */
public class SlidingWindowDemo {
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // env.setParallelism(1);
        final DataStreamSource<String> stream = env.socketTextStream("aliyun", 7777, "\n");

        final SingleOutputStreamOperator<Tuple2<String, Integer>> mapStream = stream.map(new MapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String word) throws Exception {

                final int random = new Random().nextInt(10);
                final long now = System.currentTimeMillis();
                SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");

                System.out.println("value : " + word + " random : " + random + " timestamp : " + now + "|" + format.format(now));

                return new Tuple2<>(word, random);
            }
        });

        final KeyedStream<Tuple2<String, Integer>, String> keyedStream = mapStream.keyBy(tuple -> tuple.f0);

        // 基于时间
        final WindowedStream<Tuple2<String, Integer>, String, TimeWindow> timeWindow = keyedStream.timeWindow(Time.seconds(10), Time.seconds(5));
        timeWindow.sum(1).print();

        // 基于事件
        final WindowedStream<Tuple2<String, Integer>, String, GlobalWindow> countWindow = keyedStream.countWindow(3, 2);
        countWindow.sum(1).print();

        env.execute();
    }
}
